package com.bw.fun;

import com.alibaba.fastjson.JSON;
import com.bw.bean.OrderDetailStats;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Collections;

public class AsyncIOFunc extends RichAsyncFunction<OrderDetailStats,OrderDetailStats> {
    Connection conn = null;
    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        conn = DriverManager.getConnection("jdbc:phoenix:hadoop102:2181");
        conn.setAutoCommit(true);
    }
    @Override
    public void close() throws Exception {
        conn.close();
    }

    @Override
    public void asyncInvoke(OrderDetailStats orderDetailStats, ResultFuture<OrderDetailStats> resultFuture) throws Exception {
        PreparedStatement ps = conn.prepareStatement("select * from tms.dim_base_region_info where id=?" );
        ps.setString(1,orderDetailStats.getSender_province_id());
        ResultSet rs = ps.executeQuery();
        if(rs.next()){
            String line = rs.getString("data");
            String name = JSON.parseObject(line).getString("name");
            orderDetailStats.setSender_province_name(name);
        }

        resultFuture.complete(Collections.singleton(orderDetailStats));
    }
}
